package com.shujia.flink.tf;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class Demo08Process {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> ds01 = env.socketTextStream("master", 8888);

        ds01.process(new ProcessFunction<String, Object>() {
            /*
             * 每进来一条数据就会执行一次
             * value ：一条数据
             * ctx：可以获取任务执行时的信息
             * out：用于输出数据
             */
            @Override
            public void processElement(String value, ProcessFunction<String, Object>.Context ctx, Collector<Object> out) throws Exception {
                // 通过processElement实现Map算子操作
                out.collect(Tuple2.of(value, 1));
                // 通过processElement实现flatMap算子操作
                for (String word : value.split(",")) {
                    out.collect(word);
                }
                // 通过processElement实现filter算子操作
                if("java".equals(value)){
                    out.collect("java ok");
                }
            }
        }).print();


        env.execute();

    }
}
